// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/btcturk/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/btcturk/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::btcturk::feed_v1::impl {

void BtcturkFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  // Possible update messages:
  // [432,{"CS":162688,"PS":"ATOMTRY","AO":[{"CP":3,"P":"358.49","A":"25"},{"CP":3,"P":"359.77",
  // "A":"359.64"},{"CP":1,"P":"359.99","A":"359.64"},{"CP":1,"P":"372","A":"462.31"}],"BO":[
  // {"CP":0,"P":"356.87","A":"791.34"},{"CP":0,"P":"356.85","A":"15.16"}],"channel":"obdiff",
  // "event":"ATOMTRY","type":432}]
  const auto& data = packet.payload;

  rapidjson::Document document;
  auto pos = data.find_first_of('{');
  if (pos == std::string::npos) {
    LOG(INFO) << "Not target JSON message: " << packet.payload;
    return;
  }

  std::string json_str;
  if (pos == 0) {
    // rest api response, foramt is json
    json_str = packet.payload;
  } else {
    json_str = packet.payload.substr(pos, data.size() - pos - 1);
  }
  document.Parse(json_str.c_str());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError())
            << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  const auto& channel = BtcturkChannel::Parse(document);
  if (!CheckChannelFilter(channel.channel)) {
    return;
  }

  upd_mgr_.onBeginPacket(packet);
  const auto& native_symbol = channel.native_symbol;
  if (channel.channel == "trade") {
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    Trade* trade = upd_mgr_.GetTradeMutable();
    trade->Clear();
    BtcturkFeedUpdateParser::ParseTrade(document, trade);
    upd_mgr_.PublishTrade(native_symbol, false);
  } else if (channel.channel == "obdiff" && channel.type == 432) {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      BtcturkFeedUpdateParser::ParseUpdate(
          book,
          document,
          packet.timestamp);
      upd_mgr_.PublishBook();
    }
  } else if (channel.channel == "obdiff" && channel.type == 431) {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      BtcturkFeedUpdateParser::ParseSnapshot(
          book,
          document,
          packet.timestamp);
      upd_mgr_.PublishBook();
    }
  } else if (StringStartsWith(channel.channel, "?symbol={}&resolution")) {
    // kline_rest
    const std::string& native_symbol = channel.native_symbol;
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      return;
    }

    auto interval = BtcturkFeedUpdateParser::ParseKlineInterval(channel.interval);
    auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (kline != nullptr) {
      const auto& open_timestamp = document["data"]["t"].GetArray();
      const auto& open = document["data"]["o"].GetArray();
      const auto& high = document["data"]["h"].GetArray();
      const auto& low = document["data"]["l"].GetArray();
      const auto& close = document["data"]["c"].GetArray();
      const auto& volume = document["data"]["v"].GetArray();

      std::optional<int64_t> first_open_timestamp;
      if (open_timestamp.Size() > 0) {
        first_open_timestamp = open_timestamp[0].GetInt64() * 1'000'000'000LL;;
      }

      if (kline->Size() == 0) {
        kline->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < kline->RBegin()->open_timestamp) {
        kline->SnapshotClear();
      } else {
        kline->UpdateClear();
      }

      bool to_publish = false;
      for (int idx = 0; idx < open_timestamp.Size(); idx++) {
        bool freezed = BtcturkFeedUpdateParser::Parse1Kline(
            open_timestamp[idx], open[idx], high[idx], low[idx], close[idx],
            volume[idx], kline, packet.timestamp, interval);
        to_publish = to_publish || freezed;
      }

      if (to_publish) {
        upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    LOG(INFO) << "[Btcturk Feed Processor] unknown msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::btcturk::feed_v1::impl
